/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.curator.framework.recipes.locks;

import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * 标准锁内部驱动
 */
public class StandardLockInternalsDriver implements LockInternalsDriver {
    static private final Logger log = LoggerFactory.getLogger(StandardLockInternalsDriver.class);

    @Override
    public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName,
                                        int maxLeases) throws Exception {
        // 获取当前节点在自节点中的顺序
        int ourIndex = children.indexOf(sequenceNodeName);
        // 校验之前创建的临时顺序节点是否有效
        validateOurIndex(sequenceNodeName, ourIndex);

        // maxLeases默认为1 ，当ourIndex=0时为true，表示当前节点为最小节点
        // 锁公平性的核心逻辑
        // 由InterProcessMutex的构造函数可知，maxLeases为1，即只有ourIndex为0时，线程才能持有锁，或者说该线程创建的临时顺序节点激活了锁
        // Zookeeper的临时顺序节点特性能保证跨多个JVM的线程并发创建节点时的顺序性，越早创建临时顺序节点成功的线程会更早地激活锁或获得锁
        boolean getsTheLock = ourIndex < maxLeases;

        // 如果已经获得了锁，则无需监听任何节点，否则需要监听上一顺序节点（ourIndex-1）
        // 因为锁是公平的，因此无需监听除了（ourIndex-1）以外的所有节点，这是为了减少羊群效应，非常巧妙的设计！！
        String pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);

        // 返回获取锁的结果，交由上层继续处理（添加监听等操作）
        return new PredicateResults(pathToWatch, getsTheLock);
    }

    @Override
    public String fixForSorting(String str, String lockName) {
        return standardFixForSorting(str, lockName);
    }

    /**
     * 字符串处理，获取lock-后生成的序号
     * @param str   全路径 /test/lock-111
     * @param lockName  锁前缀
     */
    public static String standardFixForSorting(String str, String lockName) {
        int index = str.lastIndexOf(lockName);
        if (index >= 0) {
            System.out.println(index);
            index += lockName.length();
            return index <= str.length() ? str.substring(index) : "";
        }
        return str;
    }

    static void validateOurIndex(String sequenceNodeName, int ourIndex) throws KeeperException {
        // 容错处理，可跳过
        // 由于会话过期或连接丢失等原因，该线程创建的临时顺序节点被Zookeeper服务端删除，往外抛出NoNodeException
        // 如果在重试策略允许范围内，则进行重新尝试获取锁，这会重新重新生成临时顺序节点
        if (ourIndex < 0) {
            log.error("Sequential path not found: " + sequenceNodeName);
            throw new KeeperException.NoNodeException("Sequential path not found: " + sequenceNodeName);
        }
    }

    public static void main(String[] args) {
        String s = standardFixForSorting("/test/lock-111", "lock-");
        System.out.println(s);
    }
}
